package com.crazyice.lee.jobs;

import com.crazyice.lee.data.WordIsCount;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.regex.Pattern;

@Slf4j
public class WordCountSQL {
    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try(InputStream inputStream = WordCountSQL.class.getResourceAsStream("/file.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) throws Exception {
        init();

        // set up execution environment
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        final BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
        //数据源
        DataSource<String> logs=env.readTextFile(properties.getProperty("source")).name("从文件读取数据");

        //数据处理
        FlatMapOperator<String, WordIsCount> input = logs.flatMap(new FlatMapFunction<String, WordIsCount>() {
            public void flatMap(String value, Collector<WordIsCount> out) throws Exception {
                String[] splits = value.split("\\s");
                for (String word:splits) {
                    boolean isMatch = Pattern.matches("^[\\u4E00-\\u9FA5A-Za-z0-9_.]+$", word);
                    if(isMatch) {
                        out.collect(new WordIsCount(word, 1L));
                    }
                }
            }
        }).name("转化为POJO");

        //register the DataSet as table "WordCount"
        tEnv.registerDataSet("WordCount", input, "word, frequency");

        // run a SQL query on the Table and retrieve the result as a new Table
        Table table = tEnv.sqlQuery(
                "SELECT word, SUM(frequency) as frequency FROM WordCount GROUP BY word").orderBy("frequency.desc");
        DataSet<WordIsCount> result = tEnv.toDataSet(table, WordIsCount.class);

        //结果输出
        result.writeAsText(properties.getProperty("sink"), FileSystem.WriteMode.OVERWRITE).name("输出到文件");
        env.execute("SQL计算");
    }
}